package com.lxl.hadoop.friends;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.*;

/**
 * @Author lixilin
 * @Date 2022/1/13 21:52
 * @Description
 */
public class FriendsReducer extends Reducer<Text,Text,Text,Text> {

    Map<String, String> map = new HashMap<String, String>();
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            map.put(key.toString(),value.toString());
        }
    }

    Text k = new Text();
    Text v = new Text();
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        List<String> keyList = new ArrayList<String>(map.keySet());
        int size = keyList.size();
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < size; i++) {
            String f1 = keyList.get(i);
            String[] arr1 = map.get(f1).split(",");
            List<String> f1List =new ArrayList<String>(arr1.length);
            Collections.addAll(f1List, arr1);
            for (int j = i+1; j < size; j++) {
                String f2 = keyList.get(j);
                String[] arr2 = map.get(f2).split(",");
                List<String> f2List = new ArrayList<String>(arr2.length);
                Collections.addAll(f2List, arr2);
                f2List.retainAll(f1List);
                if (f2List.size() > 0) {
                    sb.setLength(0);
                    sb.append(f1);
                    sb.append("-");
                    sb.append(f2);
                    k.set(sb.toString());
                    v.set(f2List.toString());
                    context.write(k,v);
                }
            }
        }
    }
}
